import requests
from flask import Flask
import json
import pika
import threading
import logging
from similar_question_process import insert_sim_question, delete_sim_question, update_sim_question
from file_content_process import update_file_content, delete_file_content

# 创建logger对象
logger = logging.getLogger('mylogger')
logger.setLevel(logging.INFO)

# 创建FileHandler对象
fh = logging.FileHandler('logs/mq_log.log')
fh.setLevel(logging.INFO)

ch = logging.StreamHandler()
ch.setLevel(logging.INFO)

# 创建Formatter对象
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
ch.setFormatter(formatter)

# 将FileHandler对象添加到Logger对象中
logger.addHandler(fh)
logger.addHandler(ch)


app = Flask(__name__)


class RmqDB:
    def __init__(self, username, password, host, port, queue="QA_QUEUE"):
        self.username = username
        self.password = password
        self.host = host
        self.queue = queue
        self.port = port
        self.connection = None
        self.channel = None
        self.connect()

    def connect(self):
        try:
            user_pwd = pika.PlainCredentials(self.username, self.password)
            self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host,
                                                                                port=self.port,
                                                                                credentials=user_pwd,
                                                                                heartbeat=10,
                                                                                socket_timeout=10,
                                                                                stack_timeout=20,
                                                                                retry_delay=5,
                                                                            ))
            self.channel = self.connection.channel()
            self.channel.queue_declare(queue=self.queue, durable=True)
        except Exception as e:
            logger.error(f"ConnectError: {e}")


    def publish(self, item):
        items = json.dumps(item)
        # if not self.channel or self.channel.is_closed:
        #     try:
        #         self.connection.close()
        #     finally:
        #         self.connect()
        self.channel.basic_publish(
            exchange='',
            routing_key=self.queue,
            body=items)

    def consume(self):
        if not self.channel or self.channel.is_closed:
            try:
                self.connection.close()
            except Exception as e:
                print(f"ConnectClose: {e}")
            finally:
                self.connect()
        self.channel.basic_consume(queue=self.queue, on_message_callback=self._callback, auto_ack=False)
        try:
            self.channel.start_consuming()
        except pika.exceptions.ConnectionClosedByBroker as e:
            logger.error(f"ConnectionClosedByBroker: {e}")
        except pika.exceptions.AMQPChannelError as e:
            logger.error(f"AMQPChannelError: {e}")
        except pika.exceptions.AMQPConnectionError as e:
            logger.error(f"AMQPConnectionError: {e}")
        except Exception as e:
            logger.error(f"OtherError: {e}")
        # self.connection.close()
        self.consume()

    def _callback(self, ch, method, properties, body):
        content_encoding = properties.headers.get('content_encoding', 'utf-8')
        body = body.decode(content_encoding)
        logger.info("message:" + body)
        try:
            body = json.loads(body)
            qa_type = body["messageType"]
            data = body["data"]
            opType = data["opType"]
            res_message = {"code": 400, "message": "Undefined operation!"}
            if qa_type == "FAQ":
                if opType == "ADD":
                    res_message = insert_sim_question(data, logger)
                if opType == "UPDATE":
                    res_message = update_sim_question(data, logger)
                if opType == "DELETE":
                    res_message = delete_sim_question(data, logger)
                if opType == "INIT":
                    res_message = {"code": 200}
            elif qa_type == "DOC":
                if opType == "ADD":
                    res_message = update_file_content(data, logger)
                if opType == "UPDATE":
                    res_message = update_file_content(data, logger)
                if opType == "DELETE":
                    res_message = delete_file_content(data, logger)
                if opType == "INIT":
                    res_message = {"code": 200}
        except Exception as e:
            res = requests.post("http://113.31.111.86:19003/knowledge1/save_error_data/", data={
                "message": str(body),
                "error_info": str(e)
            })
            logger.error("error data save success!")
            logger.error("error:" + str(e))
            res_message = {"code": 400, "message": "data error!"}
        logger.info("results:" + str(res_message))

        ch.basic_ack(delivery_tag=method.delivery_tag)

#     def _callback(self, ch, method, properties, body):
#         data = json.loads(body)
#         print(f"data: {data}, ch:{ch}, method: {method}, properties: {properties}")
#         # TODO: 处理数据
#         # insert_data(data)

#         # max_int = int(random.randint(1, 10))
#         # print(f"costTime: {max_int}\n\n\n")
#         # time.sleep(max_int)

#         # TODO: 手动删除缓存
#         ch.basic_ack(delivery_tag=method.delivery_tag)

    def close(self):
        if self.connection and self.connection.is_closed:
            self.connection.close()


def handle():
    print(f"starting")
    mqconnect = RmqDB(username='admin',
                    password='admin',
                    host='1.116.208.128',
                    port=5672)
    mqconnect.consume()
    print(f"ending")


# def listen():
#     handle()
#     app.run(host="0.0.0.0", port=2168)


if __name__ == '__main__':
    # todo: 生产测试数据
    # data_list = []
#     for i in range(20):
#         data = {'data': {'answer': f'测试{i}', 'firstGroupName': 'ZKTEST', 'opType': 'ADD', 'question': '测试4', 'question_id': 957368072610787328, 'secondGroupName': 'TEST01', 'thirdGroupName': '测试知识库'}, 'messageType': 'FAQ'}
#         data_list.append(data)
#     mqconnect = RmqDB(username='admin',
#                     password='admin',
#                     host='1.116.208.128',
#                     port=5672)
#     for i in data_list:
#         mqconnect.publish(i)

#     time.sleep(3)

    # todo: 多线程消费
    # listen()
    T_POOL = []
    # t1 = threading.Thread(target=listen)
    # T_POOL.append(t1)
    print("THREAD:", 20)
    for i in range(20):
        print(f"i:{i}")
        t = threading.Thread(target=handle)
        T_POOL.append(t)

    for i in T_POOL:
        i.start()

    for i in T_POOL:
        i.join()

